package cn.com.pc.content.kafka.consumer;


import cn.com.pc.content.domain.content.ContentType;
import cn.com.pc.content.domain.article.Article;
import cn.com.pc.content.domain.customizedContent.CustomizedContent;
import cn.com.pc.content.domain.post.Post;
import cn.com.pc.content.domain.video.Video;
import cn.com.pc.content.dto.ArticleDTO;
import cn.com.pc.content.dto.CustomizedContentDTO;
import cn.com.pc.content.dto.PostDTO;
import cn.com.pc.content.dto.VideoDTO;
import cn.com.pc.content.kafka.producer.ContentProducer;
import cn.com.pc.content.service.ContentConsumerService;
import cn.com.pc.content.util.JsonUtils;
import com.fasterxml.jackson.core.type.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Optional;

@Component
@Slf4j
public class ContentConsumer {
    @Autowired
    private ContentConsumerService contentConsumerService;

    @Autowired
    private ContentProducer contentProducer;

    @KafkaListener(topics = {"pcp-content"},
            autoStartup = "${rebuildEsIndex.isKafkaConsumerStart:true}",
            containerFactory = "consumerFactory")
    public void consumer(ConsumerRecord<String, String> consumerRecord, Acknowledgment ack) {
        Map<String, Object> messageMap = null;
        try {
            Optional<String> kafkaMsg = Optional.ofNullable(consumerRecord.value());
//            log.info(">>>>>>>>>>kafka record :{}", consumerRecord);
            if (!kafkaMsg.isPresent()) {
                log.error("kafkaMsg is null; kafkaMsg key :{}", consumerRecord.key());
                return;
            }

            String message = kafkaMsg.get();
            messageMap = JsonUtils.jsonToObj(message, new TypeReference<Map<String, Object>>() {
            });
            ContentType contentType = contentConsumerService.validateKeyAndGetContentType(String.valueOf(consumerRecord.key()));
            String action = null;
            if (messageMap != null) {
                action = String.valueOf(messageMap.get("action"));
            }

            switch (contentType) {
                case Article:
                    //数据验证，无效数据发送到errorTopic
                    ArticleDTO articleDTO = (ArticleDTO) contentConsumerService.validateMsg(
                            consumerRecord.key(), message, messageMap, contentType);
                    Article article = new Article();
                    //数据处理，物料入库mongoDB和es
                    contentConsumerService.handleContentToMongoAndEs(action, articleDTO, article, contentType);
                    break;

                case Video:
                    VideoDTO videoDTO = (VideoDTO) contentConsumerService.validateMsg(
                            consumerRecord.key(), message, messageMap, contentType);
                    Video video = new Video();
                    contentConsumerService.handleContentToMongoAndEs(action, videoDTO, video, contentType);
                    break;
                case Post:
                    PostDTO postDTO = (PostDTO) contentConsumerService.validateMsg(
                            consumerRecord.key(), message, messageMap, contentType);
                    Post post = new Post();
                    contentConsumerService.handleContentToMongoAndEs(action, postDTO, post, contentType);
                    break;
                case Question:
                case Audio:
                case Course:
                case ShortMessage:
                    CustomizedContentDTO customizedContentDTO = (CustomizedContentDTO) contentConsumerService.validateMsg(
                            consumerRecord.key(), message, messageMap, contentType);
                    customizedContentDTO.setContentType(contentType);
                    CustomizedContent customizedContent = new CustomizedContent();
                    contentConsumerService.handleContentToMongoAndEs(action, customizedContentDTO, customizedContent, contentType);
                    break;

//                    todo 待增加实体类
//                case AD:
//                    validError = true;
//                    break;
//                case Product:
//                    validError = true;
//                    break;
//                case Other:
//                    validError = true;
//                    break;
                default:
                    throw new ConsumerException("kafka key contentType is error!");
            }

        } catch (Exception e) {
            if (e instanceof ConsumerException) {
                log.debug("kafka物料消费异常 -key({}): {}", consumerRecord.key(), e.getMessage());
            } else {
                log.error("kafka物料消费异常 -key({}): ", consumerRecord.key(), e);
            }

            if (messageMap != null) {
                messageMap.put("errorReason", e.getMessage());
            }
            contentProducer.sendToErrorTopic(consumerRecord.key(), JsonUtils.obj2String(messageMap));
        } finally {
            ack.acknowledge();
        }

    }


}
